package com.etlmaster.rowprocessor.core;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.executor.utils.MultiProcessHelper;
import com.etlmaster.rowprocessor.disruptor.readEvent.LineEvent;
import com.etlmaster.rowprocessor.disruptor.readEvent.LineEventFactory;
import com.etlmaster.rowprocessor.utils.ReflectionUtil;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class Job {
	private AbstractReader reader;
	private List<Processor> processors = new ArrayList<Processor>();
	private List<AbstractWriter> writers = new ArrayList<AbstractWriter>();
	private AbstractErrorPlugin errorHandler;
	private String speedInfo;
	private Map<String, Object> variables = new HashMap<String, Object>();
	public static final int NOT_STARTED = -10, // 未启动
			ABORTED = -1, // 异常中止
			RUNNING = 10, // 运行中
			SUSPENDED = 11; // 挂起
	private int status = NOT_STARTED;
	private String id;
	private String json;
	private Monitor monitor = new Monitor();
	private Disruptor<LineEvent> errorDiruptor;
	private RingBuffer<LineEvent> errorringBuffer;

	private List<Map<String, Object>> relations;
	public Job(String id,String json) {
		this.setId(id);
		this.setJson(json);
	}
	public void initErrorDisruptor(){
		LineEventFactory factory = new LineEventFactory();
		int bufferSize = 1024*16;
		errorDiruptor = new Disruptor<>(factory, bufferSize,
				MultiProcessHelper.executor);
		errorDiruptor.handleEventsWith(getErrorHandler());
		errorDiruptor.start();
		setErrorringBuffer(errorDiruptor.getRingBuffer());
	}
	public void initJob(Map<String, Object> jobParam){
		initVars((Map<String, Object>) jobParam.get("vars"));
		parseJob((List<Map<String, Object>>) jobParam.get("steps"));

		relations=(List<Map<String, Object>>) jobParam
				.get("relations");
		

		String isDebug = (String) getVariables().get("isDebug");
		if (isDebug != null && isDebug.equalsIgnoreCase("true"))
			LogWriter.initLogger(id, true);
		else
			LogWriter.initLogger(id, false);
		initPlugins();
	}

	private void parseJob(List<Map<String, Object>> jobConf) {
		for (Map<String, Object> map : jobConf) {
			String pluginType = (String) map.get("type");
			switch (pluginType) {
			case "reader":
				reader = (AbstractReader) setPlugin(map);
				break;
			case "writer":
				AbstractWriter writer = (AbstractWriter) setPlugin(map);

				getWriters().add(writer);
				break;
			case "processor":
				Processor processor = (Processor) setPlugin(map);

				getProcessors().add(processor);
				break;
			case "errorHandler":
				setErrorHandler((AbstractErrorPlugin)setPlugin(map));
				getMonitor().getSpeedInfo().put(getErrorHandler().getId(), new SpeedMeta((long) 0));

				break;
			}
		}
	}
	public void start(){
		LogWriter.addLog(this.getId(),0,"INFO", "jobId 开始启动:{}", id);
		startDisruptor(relations);
		startSpeedMonitor(this);
		LogWriter.addLog(this.getId(),0,"INFO", "jobId {}启动完成", id);
		try {
			reader.write();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public void startSpeedMonitor(final Job job){

		Runnable runnable=new Runnable(){
			public void run(){
				LogWriter.addLog("INFO",monitor.toString(job));
			}
		};
        ScheduledExecutorService service = Executors    
                .newSingleThreadScheduledExecutor();    
        // 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间    
        service.scheduleAtFixedRate(runnable, 10, 10, TimeUnit.SECONDS);    
	}
	public void startDisruptor(List<Map<String, Object>> relations) {
		for (Map<String, Object> relation : relations) {
			String sourceId = (String) relation.get("source");
			String[] targetIds = ((String) relation.get("target")).split(",");
			AbstractInputPlugin[] handlers = new AbstractInputPlugin[targetIds.length];
			AbstractOutPutPlugin outPutPlugin = (AbstractOutPutPlugin) getPluginById(sourceId);
			for (int i = 0; i < targetIds.length; i++) {
				String targetId = targetIds[i];
				handlers[i] = (AbstractInputPlugin) getPluginById(targetId);
			}
			initDisruptor(outPutPlugin, handlers);
		}
		initErrorDisruptor();
		
	}
	public AbstractPlugin getPluginById(String id) {
		AbstractPlugin plugin = null;
		if (reader.getId().equalsIgnoreCase(id))
			plugin = reader;
		else {
			for (Processor processor : this.processors) {
				if (processor.getId().equalsIgnoreCase(id))
					plugin = processor;
			}
			for (AbstractWriter writer : this.writers) {
				if (writer.getId().equalsIgnoreCase(id))
					plugin = writer;
			}
		}
		getMonitor().getSpeedInfo().put(id, new SpeedMeta((long) 0));
		return plugin;
	}

	public void stop() {
		LogWriter.addLog(this.getId(),0,"INFO", "jobId 开始停止:{}", id);
		if (reader != null) {
			LogWriter.addLog(this.getId(),0,"INFO", "停止jobId:{},reader:{}", this.id,
					reader.getId());
			reader.stop();
			reader.getDisruptor().shutdown();
			LogWriter.addLog(this.getId(),0,"INFO", "jobId:{},reader:{}停止完成", this.id,
					reader.getId());
		}
		if (processors != null) {
			for (Processor process : processors) {
				LogWriter.addLog(this.getId(),0,"INFO", "停止jobId:{},processor:{}", this.id,
						process.getId());
				process.stop();
				process.getDisruptor().shutdown();
				LogWriter.addLog(this.getId(),0,"INFO", "jobId:{},processor:{}停止完成", this.id,
						process.getId());
			}
		}
		if(writers!=null){
			for(AbstractWriter writer:writers){
				LogWriter.addLog(this.getId(),0,"INFO", "停止jobId:{},writer:{}", this.id,
						writer.getId());
				writer.stop();
				LogWriter.addLog(this.getId(),0,"INFO", "jobId:{},writer:{}停止完成", this.id,
						writer.getId());
			}
		}
		if(getErrorHandler()!=null){
			LogWriter.addLog(this.getId(),0,"INFO", "停止jobId:{},errorHanlder:{}", this.id,
					getErrorHandler().getId());
			errorDiruptor.shutdown();
			getErrorHandler().stop();
			LogWriter.addLog(this.getId(),0,"INFO", "jobId:{},errorHanlder:{}停止完成", this.id,
					getErrorHandler().getId());
		}
		LogWriter.addLog(this.getId(),0,"INFO", "jobId {}停止完成", id);
		LogWriter.addLog("INFO",monitor.getFinalReport(this));

	}

	public void initDisruptor(AbstractOutPutPlugin output,
			AbstractInputPlugin[] inputs) {
		LineEventFactory factory = new LineEventFactory();
		int bufferSize = 1024*16;
		Disruptor<LineEvent> disruptor = new Disruptor<>(factory, bufferSize,
				MultiProcessHelper.executor);
		disruptor.handleEventsWith(inputs);
		disruptor.start();
		RingBuffer<LineEvent> ringBuffer = disruptor.getRingBuffer();
		output.setRingBuffer(ringBuffer);
		output.setDisruptor(disruptor);
	}

	private AbstractPlugin setPlugin(Map<String, Object> map) {
		AbstractPlugin plugin = ReflectionUtil.createInstance(
				(String) map.get("id"), id, (String) map.get("name"),
				(String) map.get("code"), (String) map.get("class"),
				(Map) map.get("param"), variables);
		return plugin;

	}
	public void initPlugins(){
		if (reader != null) {
			this.reader.init();
		}
		if (processors != null) {
			for(Processor processor:processors){
				processor.init();
			}
		}
		if (writers != null) {
			for(AbstractWriter writer:writers){
				writer.init();
			}
		}
		if(getErrorHandler()!=null){
			getErrorHandler().init();
		}
	}

	public void initVars(Map<String, Object> map) {
		String taskId = DataCollectionUtils.getCurrentTimestampByFormate();
		variables.put("taskid", taskId);
		variables.put("jobid", this.getId());
		if(map!=null)
		variables.putAll(map);
		
	}

	public AbstractReader getReader() {
		return reader;
	}

	public void setReader(AbstractReader reader) {
		this.reader = reader;
	}

	public List<AbstractWriter> getWriters() {
		return writers;
	}

	public void setWriters(List<AbstractWriter> writers) {
		this.writers = writers;
	}

	public List<Processor> getProcessors() {
		return processors;
	}

	public void setProcessors(List<Processor> processors) {
		this.processors = processors;
	}

	public String getId() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

	public Monitor getMonitor() {
		return monitor;
	}

	public void setMonitor(Monitor monitor) {
		this.monitor = monitor;
	}

	public int getStatus() {
		return status;
	}

	public void setStatus(int status) {
		this.status = status;
	}

	public String getSpeedInfo() {
		return speedInfo;
	}

	public void setSpeedInfo(String speedInfo) {
		this.speedInfo = speedInfo;
	}

	public Map<String, Object> getVariables() {
		return variables;
	}

	public void setVariables(Map<String, Object> variables) {
		this.variables = variables;
	}

	public String getJson() {
		return json;
	}

	public void setJson(String json) {
		this.json = json;
	}
	public RingBuffer<LineEvent> getErrorringBuffer() {
		return errorringBuffer;
	}
	public void setErrorringBuffer(RingBuffer<LineEvent> errorringBuffer) {
		this.errorringBuffer = errorringBuffer;
	}
	public AbstractErrorPlugin getErrorHandler() {
		return errorHandler;
	}
	public void setErrorHandler(AbstractErrorPlugin errorHandler) {
		this.errorHandler = errorHandler;
	}
}
